package io.github.wanggit.antrpc.commons.generic;

import io.github.wanggit.antrpc.commons.bean.*;
import io.github.wanggit.antrpc.commons.codec.cryption.ICodec;
import io.github.wanggit.antrpc.commons.codec.cryption.NoOpCodec;
import io.github.wanggit.antrpc.commons.codec.serialize.ISerializer;
import io.github.wanggit.antrpc.commons.codec.serialize.json.JsonSerializer;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.future.ReadClientFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.HashMap;

/**
 *
 *
 * <pre>
 * 范化调用执行者，必须要在Zookeeper中进行服务提供者核对，避免大量创建Socket，导致DoS攻击
 *
 * </pre>
 */
@Slf4j
public class DefaultServiceProviderInvoker implements IServiceProviderInvoker {

    private ICodec codec;

    private final ISerializer serializer;

    private CuratorFramework curatorFramework;

    public DefaultServiceProviderInvoker(ICodec codec, String codecKey, String zkServers) {
        if (null == zkServers) {
            throw new IllegalArgumentException("zkServers can not be null.");
        }
        try {
            initCuratorFramework(zkServers);
        } catch (Exception e) {
            throw new IllegalStateException(
                    "Create DefaultServiceProviderInvoker init Zookeeper server error.", e);
        }
        this.codec = codec;
        if (null == this.codec) {
            this.codec = new NoOpCodec();
        }
        this.codec.setKey(codecKey);
        serializer = new JsonSerializer();
        serializer.setConfigs(new HashMap<>());
        serializer.init();
    }

    public DefaultServiceProviderInvoker(String zkServers) {
        this(new NoOpCodec(), null, zkServers);
    }

    /**
     * 必须要验证InvokeDTO.host在Zookeeper中能找到，
     * 避免通过DefaultClients.getInstance().getDefaultClient()方法大量创建IClient，从而产生DoS攻击
     *
     * @param invokeDTO invoke dto
     * @return result
     */
    @Override
    public Object invoke(InvokeDTO invokeDTO) {
        RpcRequestBean requestBean = createBasicRpcRequestBean();
        requestBean.setArgumentTypes(invokeDTO.getParameterTypeNames());
        requestBean.setFullClassName(invokeDTO.getInterfaceName());
        requestBean.setMethodName(invokeDTO.getMethodName());
        requestBean.setArgumentValues(invokeDTO.getArgumentValues());
        Host targetHost = invokeDTO.getHost();
        if (null == targetHost) {
            throw new IllegalArgumentException("InvokeDTO.host can not be null.");
        }
        boolean exists = checkHostOnZookeeper(targetHost);
        if (!exists) {
            throw new IllegalArgumentException(
                    targetHost.getHostInfo() + " Not Found in Zookeeper.");
        }
        RpcProtocol protocol = new RpcProtocol();
        protocol.setCmdId(IdGenHelper.getId());
        protocol.setType(ConstantValues.BIZ_TYPE);
        // 范化调用时，只能使用Json序列化反序列化
        protocol.setSerializer(ConstantValues.JSON_SERIALIZER);
        protocol.setData(serializer.serialize(requestBean));
        IClient client;
        try {
            client = DefaultClients.getInstance().getDefaultClient(targetHost, codec, serializer);
        } catch (InterruptedException e) {
            if (log.isErrorEnabled()) {
                log.error("Create DefaultClient ERROR.", e);
            }
            Thread.currentThread().interrupt();
            return null;
        }
        ReadClientFuture future = client.send(protocol, invokeDTO.getTimeoutSeconds());
        RpcResponseBean responseBean = future.get();
        if (null == responseBean) {
            throw new TimeoutException(invokeDTO);
        }
        return responseBean.getResult();
    }

    private boolean checkHostOnZookeeper(Host targetHost) {
        String path = "/" + ConstantValues.ZK_ROOT_NODE_NAME + "/" + targetHost.getHostInfo();
        try {
            return null != curatorFramework.checkExists().forPath(path);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Check " + path + " exists?. error.", e);
            }
            return false;
        }
    }

    private RpcRequestBean createBasicRpcRequestBean() {
        RpcRequestBean rpcRequestBean = new RpcRequestBean();
        rpcRequestBean.setTs(System.currentTimeMillis());
        rpcRequestBean.setOneway(false);
        rpcRequestBean.setCaller(ConstantValues.GENERIC_CALLER);
        rpcRequestBean.setId(SerialNumberThreadLocal.CALLER_ROOT);
        rpcRequestBean.setSerialNumber(IdGenHelper.getUUID());
        return rpcRequestBean;
    }

    private void initCuratorFramework(String zkServers) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5, 30000);
        curatorFramework =
                CuratorFrameworkFactory.builder()
                        .connectString(zkServers)
                        .connectionTimeoutMs(5000)
                        .retryPolicy(retryPolicy)
                        .build();
        curatorFramework.start();
    }
}
